package com.example.mq1.mqserver;

import com.example.mq1.common.BinaryTool;
import com.example.mq1.common.Consumer;
import com.example.mq1.common.MqException;
import com.example.mq1.mqserver.core.BasicProperties;
import com.example.mq1.network.*;

import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

//public class BrokerServer {
//
//    private ServerSocket serverSocket = null;
//    //这里考虑一个 ServerSocket 只有一个 VirtualHost
//    private VirtualHost virtualHost = new VirtualHost("virtualHost1");
//    private ConcurrentHashMap<String, Socket> sessions = new ConcurrentHashMap<>();
//    private ExecutorService pool = null;
//    private volatile boolean runnable = true;
//
//    public BrokerServer(int port) throws IOException {
//        serverSocket = new ServerSocket(port);
//    }
//
//    public void start() throws IOException {
//        System.out.println("BrokerServer 启动！");
//        pool = Executors.newCachedThreadPool();
//        try {
//            while(runnable) {
//                //处理一个客户端 socket 连接
//                Socket clientSocket = serverSocket.accept();
//                pool.submit(() -> {
//                    processConnection(clientSocket);
//                });
//            }
//        } catch (SocketException e) {
//            System.out.println("[BrokerServer] 服务器正常停止！");
//        }
//    }
//
//    public void stop() throws IOException {
//        runnable = false;
//        pool.shutdown();
//        serverSocket.close();
//    }
//
//    private void processConnection(Socket clientSocket) {
//        try (InputStream inputStream = clientSocket.getInputStream();
//             OutputStream outputStream = clientSocket.getOutputStream()) {
//            try (DataInputStream dataInputStream = new DataInputStream(inputStream);
//                 DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
//                while (true) {
//                    //1.读取请求并解析
//                    Request request = readRequest(dataInputStream);
//                    //2.根据请求计算响应
//                    Response response = process(request, clientSocket);
//                    //3.将请求写回客户端
//                    writeResponse(dataOutputStream, response);
//                }
//            }
//        } catch (EOFException | SocketException e) {
//            //这里文件读取到末尾(EOF)或者断开连接(测试完后断开连接)就会引发异常(顺水推舟)
//            System.out.println("[BrokerServer] connection 关闭！ 客户端地址: " + clientSocket.getInetAddress() +
//                    ":" + clientSocket.getPort());
//        } catch (IOException | MqException | ClassNotFoundException e) {
//            System.out.println("[BrokerServer] connection 连接异常！");
//            throw new RuntimeException(e);
//        } finally {
//            try {
//                clientSocket.close();
//                //清理当前 socket 连接中所有的 session 信息
//                clearSessions(clientSocket);
//            } catch (IOException e) {
//                throw new RuntimeException(e);
//            }
//        }
//    }
//
//
//    private Request readRequest(DataInputStream dataInputStream) throws IOException, MqException {
//        Request request = new Request();
//        request.setType(dataInputStream.readInt());
//        request.setLength(dataInputStream.readInt());
//        byte[] payload = new byte[request.getLength()];
//        int n = dataInputStream.read(payload);
//        if(n != request.getLength()) {
//            throw new MqException("[BrokerServer] 读取请求格式出错！ type=" + request.getType() +
//                    ", expectedLength=" + request.getLength() + ", actualLength=" + n);
//        }
//        request.setPayload(payload);
//        return request;
//    }
//
//    private void writeResponse(DataOutputStream dataOutputStream, Response response) throws IOException {
//        dataOutputStream.writeInt(response.getType());
//        dataOutputStream.writeInt(response.getLength());
//        dataOutputStream.write(response.getPayload());
//        dataOutputStream.flush();
//    }
//
//    private Response process(Request request, Socket clientSocket) throws IOException, ClassNotFoundException, MqException {
//        //1.先做个初步解析
//        BasicRequest basicRequest = (BasicRequest) BinaryTool.fromBytes(request.getPayload());
//        System.out.println("[Request] rid=" + basicRequest.getRid() + ", channelId=" + basicRequest.getChannelId() +
//                ", type=" + request.getType() + ", length=" + request.getLength());
//        boolean ok = true;
//        if(request.getType() == 0x1) {
//            //创建 Channel
//            sessions.put(basicRequest.getChannelId(), clientSocket);
//            System.out.println("[BrokerServer] channel 创建完成！ channelId=" + basicRequest.getChannelId());
//        } else if(request.getType() == 0x2) {
//            //删除 Channel
//            sessions.remove(basicRequest.getChannelId());
//            System.out.println("[BrokerServer] channel 删除完成！ channelId=" + basicRequest.getChannelId());
//        } else if(request.getType() == 0x3) {
//            //创建交换机
//            ExchangeDeclareArguments exchangeDeclareArguments = (ExchangeDeclareArguments) basicRequest;
//            ok = virtualHost.exchangeDeclare(exchangeDeclareArguments.getExchangeName(), exchangeDeclareArguments.getExchangeType(),
//                    exchangeDeclareArguments.isDurable(), exchangeDeclareArguments.isAutoDelete(), exchangeDeclareArguments.getArguments());
//        } else if(request.getType() == 0x4) {
//            //删除交换机
//            ExchangeDeleteArguments exchangeDeleteArguments = (ExchangeDeleteArguments) basicRequest;
//            ok = virtualHost.exchangeDelete(exchangeDeleteArguments.getExchangeName());
//        } else if(request.getType() == 0x5) {
//            //创建队列
//            QueueDeclareArguments queueDeclareArguments = (QueueDeclareArguments) basicRequest;
//            ok = virtualHost.queueDeclare(queueDeclareArguments.getQueueName(), queueDeclareArguments.isDurable(),
//                    queueDeclareArguments.isExclusive(), queueDeclareArguments.isAutoDelete(), queueDeclareArguments.getArguments());
//        } else if(request.getType() == 0x6) {
//            //删除队列
//            QueueDeleteArguments queueDeleteArguments = (QueueDeleteArguments) basicRequest;
//            ok = virtualHost.queueDelete(queueDeleteArguments.getQueueName());
//        } else if(request.getType() == 0x7) {
//            //创建绑定
//            QueueBindArguments queueBindArguments = (QueueBindArguments) basicRequest;
//            ok = virtualHost.queueBind(queueBindArguments.getExchangeName(), queueBindArguments.getQueueName(),
//                    queueBindArguments.getBindingKey());
//        } else if(request.getType() == 0x8) {
//            //删除绑定
//            QueueUnBindArguments queueUnBindArguments = (QueueUnBindArguments) basicRequest;
//            ok = virtualHost.queueUnBind(queueUnBindArguments.getExchangeName(), queueUnBindArguments.getQueueName());
//        } else if(request.getType() == 0x9) {
//            //发布消息
//            BasicPublishArguments basicPublishArguments = (BasicPublishArguments) basicRequest;
//            ok = virtualHost.basicPublish(basicPublishArguments.getExchangeName(), basicPublishArguments.getRoutingKey(),
//                    basicPublishArguments.getBasicProperties(), basicPublishArguments.getBody());
//        } else if(request.getType() == 0xa) {
//            //订阅消息
//            BasicConsumeArguments basicConsumeArguments = (BasicConsumeArguments) basicRequest;
//            ok = virtualHost.basicConsume(basicConsumeArguments.getConsumerTag(), basicConsumeArguments.getQueueName(),
//                    basicConsumeArguments.isAutoAck(), new Consumer() {
//                        // 这个回调函数要做的就是，把服务器这边的消息发送给客户端那边
//                        @Override
//                        public void handlerDelivery(String consumerTag, BasicProperties basicProperties, byte[] body) throws IOException, MqException {
//                            //1.根据 channel(就是 consumerTag) 找到客户端 socket
//                            Socket clientSocket = sessions.get(consumerTag);
//                            if(clientSocket == null || clientSocket.isClosed()) {
//                                throw new MqException("[BrokerServer] 订阅消息的客户端已经关闭");
//                            }
//                            //2.构造响应对象
//                            MessageCallback messageCallback = new MessageCallback();
//                            messageCallback.setRid("");//这里只有响应，没有需要对应的请求，因此不需要
//                            messageCallback.setChannelId(consumerTag);
//                            messageCallback.setConsumerTag(consumerTag);
//                            messageCallback.setBasicProperties(basicProperties);
//                            messageCallback.setBody(body);
//                            messageCallback.setOk(true);
//                            byte[] payload = BinaryTool.toBytes(messageCallback);
//
//                            Response response = new Response();
//                            response.setType(0xc);
//                            response.setLength(payload.length);
//                            response.setPayload(payload);
//                            //3.将响应写回客户端
//                            DataOutputStream dataOutputStream = new DataOutputStream(clientSocket.getOutputStream());
//                            writeResponse(dataOutputStream, response);
//                        }
//                    });
//        } else if(request.getType() == 0xb) {
//            //主动应答
//            BasicAckArguments basicAckArguments = (BasicAckArguments) basicRequest;
//            ok = virtualHost.basicAck(basicAckArguments.getQueueName(), basicAckArguments.getMessageId());
//        } else {
//            throw new MqException("[BrokerServer] 当前 type 非法！type=" + request.getType());
//        }
//        //3.构造响应
//        BasicResponse basicResponse = new BasicResponse();
//        basicResponse.setRid(basicRequest.getRid());
//        basicResponse.setChannelId(basicRequest.getChannelId());
//        basicResponse.setOk(ok);
//        byte[] payload = BinaryTool.toBytes(basicResponse);
//        Response response = new Response();
//        response.setType(request.getType());
//        response.setLength(payload.length);
//        response.setPayload(payload);
//        return response;
//    }
//
//
//    private void clearSessions(Socket clientSocket) {
//        List<String> toDeleteSockets = new ArrayList<>();
//        for(Map.Entry<String, Socket> entry : sessions.entrySet()) {
//            if(entry.getValue() == clientSocket) {
//                toDeleteSockets.add(entry.getKey());
//            }
//        }
//        for(String channelId : toDeleteSockets) {
//            sessions.remove(channelId);
//        }
//        System.out.println("[BrokerServer] 清理 session 完成! 被清理的 channelId=" + toDeleteSockets);
//    }
//
//}

public class BrokerServer {

    //服务器这边的 Socket
    private ServerSocket serverSocket = null;
    //虚拟主机连接
    private VirtualHost virtualHost = new VirtualHost("virtualHost1");
    //记录每个和服务器交互的 channel，key 是 channelId
    private ConcurrentHashMap<String, Socket> channelMap = new ConcurrentHashMap<>();
    //控制服务器断开连接
    private volatile boolean runnable = true;
    //使用线程池来处理每个客户端连接后需要处理的任务
    private ExecutorService executorService = null;

    /**
     * 构造时，指定端口号
     * @param port
     * @throws IOException
     */
    public BrokerServer(int port) throws IOException {
        serverSocket = new ServerSocket(9090);
    }

    /**
     * 启动服务器
     */
    public void start() throws IOException {
        System.out.println("[BrokerServer] 启动！");
        executorService = Executors.newCachedThreadPool();
        try {
            while(runnable) {
                //阻塞等待接收客户端的连接
                Socket clientSocket = serverSocket.accept();
                executorService.submit(() -> {
                    processConnection(clientSocket);
                });
            }
        } catch (SocketException e) {
            System.out.println("[BrokerServer] 服务器正常关闭！");
        }
    }

    /**
     * 停止服务器
     */
    public void stop() throws IOException {
        runnable = false;
        executorService.shutdown();
        serverSocket.close();
        channelMap.clear();
    }

    /**
     * 用来处理一个客户端 socket 连接(channel)
     * @param clientSocket
     */
    private void processConnection(Socket clientSocket) {
        try (InputStream inputStream = clientSocket.getInputStream();
             OutputStream outputStream = clientSocket.getOutputStream()) {
            try (DataInputStream dataInputStream = new DataInputStream(inputStream);
                 DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                while(true) {
                    //1.读取请求并解析
                    Request request = readRequest(dataInputStream);
                    //2.根据请求计算响应
                    Response response = process(request, clientSocket);
                    //3.将响应写入流，返回给客户端
                    writeResponse(response, dataOutputStream);
                }
            }
        } catch (SocketException e) {
            System.out.println("[BrokerServer] 客户端连接正常断开！");
        } catch (IOException | MqException | ClassNotFoundException e) {
            System.out.println("[BrokerServer] 客户端连接异常断开！");
            throw new RuntimeException(e);
        } finally {
            try {
                //释放 TCP 连接
                clientSocket.close();
                //清理对应的 channel 连接
                clearSessions(clientSocket);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }

    /**
     * 清除对应的 session 信息
     */
    private void clearSessions(Socket clientSocket) {
        List<String> toDeleteSessions = new ArrayList<>();
        for(Map.Entry<String, Socket> entry : channelMap.entrySet()) {
            if(entry.getValue() == clientSocket) {
                toDeleteSessions.add(entry.getKey());
            }
        }
        for(String channelId : toDeleteSessions) {
            channelMap.remove(channelId);
        }
    }


    /**
     * 读取请求
     * @param dataInputStream
     * @return
     */
    private Request readRequest(DataInputStream dataInputStream) throws IOException, MqException {
        int type = dataInputStream.readInt();
        int length = dataInputStream.readInt();
        byte[] payload = new byte[length];
        int n = dataInputStream.read(payload);
        if(n != length) {
            throw new MqException("[BrokerServer] 读取请求格式出错！expectedSize=" + length +
                    ", actualSize=" + n);
        }
        Request request = new Request();
        request.setType(type);
        request.setLength(length);
        request.setPayload(payload);
        System.out.println("[Request] type=" + request.getType() + ", length=" + request.getLength());
        return request;
    }

    /**
     * 根据请求计算响应
     * @param request
     * @param clientSocket
     * @return
     */
    private Response process(Request request, Socket clientSocket) throws IOException, ClassNotFoundException, MqException {
        //先对请求一个初步解析
        BasicRequest basicRequest = (BasicRequest) BinaryTool.fromBytes(request.getPayload());
        boolean ok = true;
        if(request.getType() == 0x1) {
            channelMap.put(basicRequest.getChannelId(), clientSocket);
            System.out.println("[BrokerServer] 创建 channel 成功！channelId=" + basicRequest.getChannelId());
        } else if(request.getType() == 0x2) {
            channelMap.remove(basicRequest.getChannelId());
            System.out.println("[BrokerServer] 销毁 channel 成功！channelId=" + basicRequest.getChannelId());
        } else if(request.getType() == 0x3) {
            ExchangeDeclareArguments arguments = (ExchangeDeclareArguments) basicRequest;
            ok = virtualHost.exchangeDeclare(arguments.getExchangeName(), arguments.getExchangeType(), arguments.isDurable(),
                    arguments.isAutoDelete(), arguments.getArguments());
        } else if(request.getType() == 0x4) {
            ExchangeDeleteArguments arguments = (ExchangeDeleteArguments) basicRequest;
            ok = virtualHost.exchangeDelete(arguments.getExchangeName());
        } else if(request.getType() == 0x5) {
            QueueDeclareArguments arguments = (QueueDeclareArguments) basicRequest;
            ok = virtualHost.queueDeclare(arguments.getQueueName(), arguments.isDurable(), arguments.isExclusive(),
                    arguments.isAutoDelete(), arguments.getArguments());
        } else if(request.getType() == 0x6) {
            QueueDeleteArguments arguments = (QueueDeleteArguments) basicRequest;
            ok = virtualHost.queueDelete(arguments.getQueueName());
        } else if(request.getType() == 0x7) {
            QueueBindArguments arguments = (QueueBindArguments) basicRequest;
            ok = virtualHost.queueBind(arguments.getExchangeName(), arguments.getQueueName(), arguments.getBindingKey());
        } else if(request.getType() == 0x8) {
            QueueUnBindArguments arguments = (QueueUnBindArguments) basicRequest;
            ok = virtualHost.queueUnBind(arguments.getExchangeName(), arguments.getQueueName());
        } else if(request.getType() == 0x9) {
            BasicPublishArguments arguments = (BasicPublishArguments) basicRequest;
            ok = virtualHost.basicPublish(arguments.getExchangeName(), arguments.getRoutingKey(), arguments.getBasicProperties(),
                    arguments.getBody());
        } else if(request.getType() == 0xa) {
            BasicConsumeArguments arguments = (BasicConsumeArguments) basicRequest;
            ok = virtualHost.basicConsume(arguments.getConsumerTag(), arguments.getQueueName(), arguments.isAutoAck(),
                    new Consumer() {
                        //用来将服务器这边收到的回调数据发送给客户端
                        @Override
                        public void handlerDelivery(String consumerTag, BasicProperties basicProperties, byte[] body) throws IOException, MqException {
                            Socket clientSocket = channelMap.get(consumerTag); //根据 channelId (这里比较特殊，客户端发来的 consumerTag 就是 channelId)来获取客户端 channel 连接
                            if(clientSocket == null || clientSocket.isClosed()) {
                                throw new MqException("[BrokerServer] 客户端关闭 channel 连接！");
                            }
                            MessageCallback messageCallback = new MessageCallback();
                            messageCallback.setRid(""); //这里不涉及请求和响应的对应关系
                            messageCallback.setChannelId(consumerTag); //这里的 channelId 实际就是 consumerTag(客户端 channel 那边发来的请求是这样约定的)
                            messageCallback.setOk(true);
                            messageCallback.setConsumerTag(consumerTag);
                            messageCallback.setBasicProperties(basicProperties);
                            messageCallback.setBody(body);
                            byte[] payload = BinaryTool.toBytes(messageCallback);

                            //给客户端传送 0xc 响应，客户端识别后，执行真正的响应
                            Response response = new Response();
                            response.setType(0xc);
                            response.setLength(payload.length);
                            response.setPayload(payload);

                            DataOutputStream dataOutputStream = new DataOutputStream(clientSocket.getOutputStream());
                            writeResponse(response, dataOutputStream);
                        }
                    });
        } else if(request.getType() == 0xb) {
            BasicAckArguments arguments = (BasicAckArguments) basicRequest;
            ok = virtualHost.basicAck(arguments.getQueueName(), arguments.getMessageId());
        } else {
            throw new MqException("[BrokerServer] 非法类型！type=" + request.getType());
        }
        //构造响应
        BasicResponse basicResponse = new BasicResponse();
        basicResponse.setRid(basicRequest.getRid());
        basicResponse.setChannelId(basicRequest.getChannelId());
        basicResponse.setOk(ok);
        byte[] payload = BinaryTool.toBytes(basicResponse);

        Response response = new Response();
        response.setType(request.getType());
        response.setLength(payload.length);
        response.setPayload(payload);
        return response;
    }

    /**
     * 将请求写回客户端
     * @param response
     * @param dataOutputStream
     */
    private void writeResponse(Response response, DataOutputStream dataOutputStream) throws IOException {
        dataOutputStream.writeInt(response.getType());
        dataOutputStream.writeInt(response.getLength());
        dataOutputStream.write(response.getPayload());
        dataOutputStream.flush();
    }


}
